package com.qax.task.timer.core;

import com.chinacreator.c2etcd.EtcdClient;
import com.chinacreator.c2etcd.EtcdClientFactory;
import com.google.common.collect.Lists;
import com.qax.task.logger.Logger;
import com.qax.task.logger.LoggerFactory;
import com.qax.task.timer.Timeout;
import com.qax.task.timer.TimerTask;
import com.qax.task.util.JsonKit;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.watch.WatchEvent;
import io.etcd.jetcd.watch.WatchResponse;
import kong.unirest.HttpResponse;
import kong.unirest.MimeTypes;
import kong.unirest.Unirest;
import lombok.Data;
import lombok.SneakyThrows;
import org.apache.commons.lang3.StringUtils;
import org.reflections.Reflections;
import org.reflections.scanners.MethodAnnotationsScanner;
import org.reflections.util.ClasspathHelper;
import org.reflections.util.ConfigurationBuilder;

import java.lang.reflect.Method;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.charset.Charset;
import java.nio.file.Path;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Supplier;

import static com.qax.task.timer.core.Constant.SLASH;
import static org.reflections.util.ClasspathHelper.forClass;

/**
 * @author lianglei
 */
@Data
public class JobSourceReporter {

    private static final Logger LOGGER = LoggerFactory.getLogger(JobSourceReporter.class);

    private static ConcurrentHashMap<String, Method> JOB_HANDLER_CACHE = new ConcurrentHashMap();

    private static Map<String, Timeout> JOB_COMPLETABLE_FUTURE_CACHE = new HashMap();

    private static ConcurrentHashMap<String, JobInfo> JOB_INFO_CACHE = new ConcurrentHashMap();

    private static ConcurrentHashMap<String, JobCommand> JOB_COMMAND_CACHE = new ConcurrentHashMap();

    public static Reflections reflections;

    public static EtcdClient etcdClient;

    /**
     * Etcd  address url ;
     */
    private String etcdAddress;

    /**
     * Upload report url for interface
     */
    private String reportInterfaceUrl;

    /**
     * Upload The job's info status url for interface
     */
    private String updateStatusRealTimeUrl;

    /**
     * Heartbeat renewal on
     */
    private boolean usedHeartBeatLease;


    private void init() {
        LOGGER.info("Reporter is starting ");
        etcdClient = EtcdClientFactory.getInstance().connect("-config",
                etcdAddress,
                usedHeartBeatLease,
                false);

        String curPackage = getClass().getPackage().getName();
        String[] t1s = curPackage.split(Constant.DOT);

        reflections =
                new org.reflections.Reflections(new ConfigurationBuilder()
                        .addUrls(forClass(Path.class))
                        .addUrls(ClasspathHelper.forPackage(t1s[0] + "." + t1s[1]))
                        .setScanners(new MethodAnnotationsScanner()));

        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Current package scanned reflections path's prefix for {} ," +
                            "etcd server address is  {}," +
                            "and the interface for job info report is {}",
                    t1s[0],
                    etcdAddress,
                    reportInterfaceUrl);
        }

        //Start to report job info mina to server of to be has assigned url
        this.doReport();
    }

    protected void doReport() {

        Set<Method> annotatedWith =
                reflections.getMethodsAnnotatedWith(JobHanlder.class);
        Iterator<Method> iterator = annotatedWith.iterator();

        ArrayList<Method> jobMethods = Lists.newArrayList(iterator);

        jobMethods.stream().forEach((method) -> {
            String jobId = UUID.randomUUID().toString();//"034d5050";

            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("The produced jobId in the time is {} ", jobId);
            }

            JobHanlder job = method.getAnnotation(JobHanlder.class);
            JobInfo jobInfo = JobInfo.builder().
                    groupName(job.groupName()).//todo 为什么没有默认值
                    applicationName(job.applicationName()).
                    jobId(jobId).
                    type(job.type()).
                    jobStatus((byte) JobStatus.INIT.getStatus()).
                    build();

            if(LOGGER.isInfoEnabled()){
                LOGGER.info("The JOB details that currently" +
                        " need to be reported to the server are is {}",jobInfo);
            }

            //Report JOB information to server storage
            URL url;
            try {
                url = new URL(reportInterfaceUrl);
            } catch (MalformedURLException e) {
                LOGGER.error("Url is invalided", e);
                return;
            }

            JOB_HANDLER_CACHE.putIfAbsent(jobId, method);

            //todo Retry strategy needs to be added here
            HttpResponse<String> response = Unirest.
                    post(url.toString()).
                    charset(Charset.defaultCharset()).
                    contentType(MimeTypes.JSON).
                    body(jobInfo).
                    asString();

            if (!response.isSuccess()) {
                LOGGER.error(response.getParsingError().get().getMessage());
                return;
            }

            //缓存jobiD对应信息
            JOB_INFO_CACHE.putIfAbsent(jobId, jobInfo);

            //3 ETCD中开启节点状态监控(/serviceType（JCQ）/serviceName(example-application)/jobGroup（默认为default,参考nacos）/jobId/status/ 0)
            StringBuilder sb = new StringBuilder();
            String statusChannelPath = sb.append(SLASH).
                    append(jobInfo.type).
                    append(SLASH).
                    append(jobInfo.applicationName).
                    append(SLASH).
                    append(jobInfo.groupName).
                    append(SLASH).
                    append(jobId/*"034d5050"*/).
                    append(SLASH).
                    append(Constant.STATUS).
                    append(SLASH).toString().intern();

            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("Initial's status channel path is {}", statusChannelPath);
            }

            try {
                //客户端创建状态节点
                {
                    etcdClient.putKey(statusChannelPath, JobStatus.INIT.toString());
                    jobInfo.jobStatus = (byte) JobStatus.INIT.getStatus();
                }

                //客户端创建TECD开启下发参数通道
                StringBuilder ssb = new StringBuilder();
                String parameterChannelPath = ssb.append(SLASH).
                        append(jobInfo.type).
                        append(SLASH).
                        append(jobInfo.applicationName).
                        append(SLASH).
                        append(jobInfo.groupName).
                        append(SLASH).
                        append(jobId/*"034d5050"*/).
                        append(SLASH).
                        append(Constant.COMMAND).
                        append(SLASH).toString().intern();

                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("Initial's parameter channel path is {}", parameterChannelPath);
                }

                // todo Async exec_
                {
                    try {  ////jcq/email-application/GROUP_NAME/034d5050/command/
                        etcdClient.putKey(parameterChannelPath, StringUtils.EMPTY);
                        if (LOGGER.isInfoEnabled()) {
                            LOGGER.info("Current listening status for path= {} and param's {}",
                                    parameterChannelPath, "");
                        }
                    } catch (Exception err) {
                        CompletableFuture.failedFuture(err);
                        LOGGER.error(err.getMessage(), err);
                        return;
                    }
                    Consumer<WatchResponse> commandConsumer = watchResponse -> {
                        Iterator<WatchEvent> iter = watchResponse.getEvents().iterator();
                        while (iter.hasNext()) {
                            WatchEvent watchEvent = iter.next();
                            //Filter delete events, because the server only monitors two events put and delete
                            if (watchEvent.getEventType().compareTo(WatchEvent.EventType.DELETE) > 0) {
                                return;
                            }
                            KeyValue keyValue = watchEvent.getKeyValue();
                            //Get the server command return value and
                            //start processing the job execution process,
                            //after the execution is complete
                            byte[] responseBytes = keyValue.getValue().getBytes();
                            JobCommand command;
                            try {
                                command = JsonKit.deserializeByteArray(responseBytes, JobCommand.class);
                                //it's cached jobCommand info for server callback ..
                                JOB_COMMAND_CACHE.put(command.jobId, command);
                                //Determine whether to execute immediately or delay execution,
                                //the component is only executed once by default.
                                this.doExecCoreSpock_(command);
                            } catch (Exception es) {
                                LOGGER.error(es.getMessage(), es);
                            }
                        }
                    };
                    if (LOGGER.isInfoEnabled()) {
                        LOGGER.info("Current's listener for parameter channel path is = {}"
                                , parameterChannelPath);
                    }
                    etcdClient.addWatcherListener(parameterChannelPath, commandConsumer);
                }

                {
                    //Open JOB return value return channel tree(不在这里写，etcd监控command参数再处理)
                    //Listening to broadcast of server endPoint
                    Consumer<WatchResponse> statusConsumer = res -> {
                        Iterator<WatchEvent> iter = res.getEvents().iterator();
                        while (iter.hasNext()) {
                            WatchEvent watchEvent = iter.next();
                            KeyValue keyValue = watchEvent.getKeyValue();
                            //Get the server command return value and start processing the job execution process,
                            //after the execution is complete
                            if (ByteSequence.from(keyValue.getValue().getBytes()).isEmpty()) {
                                return;
                            }

                            String status = new String(keyValue.getValue().getBytes());
                            JobCommand command = this.getRelatedParameterCommandAboutJobId(keyValue);
                            if (JobStatus.RUN.toString().equals(status)) {
                                //Get the commandParameter corresponding to the current JOB_ID
                                //Determine whether to execute immediately or delay execution,
                                //the component is only executed once by default
                                this.doExecCoreSpock_(command);
                            } else if (JobStatus.CANNEL.toString().equals(status)) {
                                Timeout completableFuture =
                                        JOB_COMPLETABLE_FUTURE_CACHE.get(command.jobId);
                                if (!completableFuture.isCancelled()
                                        &&
                                        !completableFuture.isExpired()) {
                                    completableFuture.cancel();
                                    //Actively delete tasks that have been completed, and take LRU for others
                                    JOB_COMPLETABLE_FUTURE_CACHE.remove(command.jobId);
                                    JobInfo info = JOB_INFO_CACHE.get(command.jobId);
                                    String statusChannelPath$ = this.fetchChannelStatusPath(command, info);
                                    try {
                                        this.syncEtcdAndNotifyEndPointStatus_(command, jobInfo, statusChannelPath$, JobStatus.CANNEL);
                                    } catch (Exception e) {
                                        LOGGER.error(e.getMessage(),e.getCause());
                                        //ignore
                                    }
                                }
                            } else if (JobStatus.INIT.toString().equals(status)) {
                                //
                                LOGGER.error("Direct operation of the channel is not allowed of Internal Error!" +
                                        " , cur status's = {}," +
                                        "the program should be initialized first", status);
                                return;
                            } else {
                                return;
                            }
                        }
                    };
                    if (LOGGER.isInfoEnabled()) {
                        LOGGER.info("Current's listener for status channel path is = {}"
                                , statusChannelPath);
                    }
                    etcdClient.addWatcherListener(statusChannelPath, statusConsumer);
                }
                //Client listening state tree node
            } catch (Exception er) {
                //ignore
                LOGGER.error(er.getMessage(), er);
                return;
            }
        });


        // TODO: 2021/2/12  ETCD开启JOB返回值回传通道树,根据服务端返回参数（isFallback），JOB执行之前创建

        //Retry tasks for processing failures,
        //which are scanned regularly by a single daemon's thread,
        //and executed once a minute by default
        Supplier supplier = () -> {
            Set<Map.Entry<String, Timeout>> entries = JOB_COMPLETABLE_FUTURE_CACHE.entrySet();
            entries.forEach((entry) -> {
                String jobId = entry.getKey();
                Timeout timeout = entry.getValue();
                JobInfo jobInfo = JOB_INFO_CACHE.get(jobId);

                if (jobInfo.jobStatus.intValue() ==
                        JobStatus.FAILURE.getStatus()) {
                    // Current timeout was also is failure in sight
                    // In the future, a customized retry mechanism is needed
                    // because the network is not trusted
                    // todo At the time , it's need to be refactored
                    try {
                        timeout.task().run(timeout);
                    } catch (Exception e) {
                        //ignore once
                        LOGGER.error(e.getMessage(), e);
                    }
                }
            });
            return null;
        };
        ScheduledCompletableFuture.scheduleFixedDelay(
                supplier,
                60,
                60,
                TimeUnit.SECONDS);

        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Initialized the fault-tolerant background thread," +
                    " the default execution delay is one minute");
        }
    }

    @SneakyThrows
    private String fetchChannelStatusPath(JobCommand command, JobInfo info) {

        StringBuilder sb$ = new StringBuilder();
        String path = sb$.append(SLASH).
                append(info.type).
                append(SLASH).
                append(info.applicationName).
                append(SLASH).
                append(info.groupName).
                append(SLASH).
                append(command.jobId/*"034d5050"*/).
                append(SLASH).
                append(Constant.STATUS).
                append(SLASH).toString().intern();

        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("The currently generated path is [{}] " +
                            ",by server parameter command  [{}] triggered ",
                    path,
                    JsonKit.serializeObject2Chars(command));
        }
        return path;
    }


    private JobCommand getRelatedParameterCommandAboutJobId(KeyValue prevKV) {
        String curJobId = this.fetchJobIdOfCurStatusChannel(prevKV);
        return Optional.
                ofNullable(JOB_COMMAND_CACHE.get(curJobId)).
                orElseThrow(NoSuchElementException::new);
    }

    private String fetchJobIdOfCurStatusChannel(KeyValue prevKV) {
        String curPath = prevKV.getKey().toString(Charset.forName("UTF-8"));
        String[] split = curPath.split(SLASH);
        return split[split.length - 2];
    }

    /**
     * <BR>Proxy method for core logic <BR/>
     * @param command
     */
    private void doExecCoreSpock_(JobCommand command) {
        Timeout completableFuture;
        if (command.triggerTime == null || command.triggerMsg.trim().isEmpty()) {
            //Immediately exec
            completableFuture = this.futureSubmitExecuted(command);
        } else {
            completableFuture = this.futureSubmitExecutedDelay(command,
                    Long.valueOf(command.triggerTime),
                    TimeUnit.MILLISECONDS);
        }
        //Global save JOB corresponds to the execution of the future, used for the global
        JOB_COMPLETABLE_FUTURE_CACHE.put(command.jobId, completableFuture);
    }

    private Timeout futureSubmitExecuted(JobCommand command) {
        TimerTask execTask = timeout -> invoke_(command);
        return TimeWheelCompletableFuture.submit(execTask);
    }


    private Timeout futureSubmitExecutedDelay(JobCommand command,
                                              long delay,
                                              TimeUnit unit) {
        TimerTask execTask = timeout -> this.invoke_(command);
        return TimeWheelCompletableFuture.schedule(
                execTask,
                delay,
                unit);
    }


    /**
     * <BR>Core execution method, responsible for specific reporting data and
     * related monitoring processes<BR/>
     * @param command Parameters issued by the dispatch center
     */
    @Async
    private void invoke_(JobCommand command) {
        Method method = JOB_HANDLER_CACHE.get(command.jobId/*"034d5050"*/);
        JobInfo jobInfo = JOB_INFO_CACHE.get(command.jobId/*"034d5050"*/);
        String statusChannelPath = this.fetchChannelStatusPath(command, jobInfo);

        try {
            // Here should be based on whether the client guides
            // the server to send execution parameters, and the SDK judges temporarily
            Object _obj = method.getDeclaringClass().getConstructor().newInstance();
            if (method.getParameterTypes().length == 0) {//No parameter method
                method.invoke(_obj, new Object[]{});
            } else {
                method.invoke(_obj, new Object[]{command.triggerMsg});
            }

            this.syncEtcdAndNotifyEndPointStatus_(
                    command,
                    jobInfo,
                    statusChannelPath,
                    JobStatus.FINISHED);

            if (LOGGER.isDebugEnabled()) {s
                LOGGER.debug("Cur's job invoke is finished by {} " +
                                "and invoke's method name is {}  " +
                                "and current's job info is {}",
                        JsonKit.serializeObject2Chars(command),
                        method.getName(),
                        JsonKit.serializeObject2Chars(jobInfo));
            }

        } catch (Exception err) {
            LOGGER.error(err.getMessage(), err);
            CompletableFuture.runAsync(() -> {
                try {
                    this.syncEtcdAndNotifyEndPointStatus_(
                            command,
                            jobInfo,
                            statusChannelPath,
                            JobStatus.FAILURE);
                } catch (Exception tr) {
                    CompletableFuture.failedFuture(tr);
                    return;
                }
            });
        }
    }

    /**
     * <BR>Synchronize configuration center and server and related status method abstract<BR/>
     * @param command
     * @param jobInfo
     * @param statusChannelPath
     * @param finished
     * @throws Exception
     */
    private void syncEtcdAndNotifyEndPointStatus_(JobCommand command, JobInfo jobInfo, String statusChannelPath, JobStatus finished) throws Exception {
        String url = String.format(
                this.updateStatusRealTimeUrl,
                finished.toString(),
                command.jobId);

        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Current updateStatusUrlRealTimeUrl is {}", url);
        }

        Unirest.put(url).asString();
        etcdClient.putKey(statusChannelPath, finished.toString());
        jobInfo.jobStatus = (byte) finished.getStatus();
    }
}


